package com.colin.util.util;

import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 自定义线程池
 * 与spring配置的线程池区别使用，避免业务较多的情况下，造成线程池阻塞
 */
public class ThreadPool {
    private static final Logger logger = LoggerFactory.getLogger(ThreadPool.class);
    /**
     * 核心线程数
     */
    private static final int corSize = 20;
    /**
     * 最大线程数
     */
    private static final int maxSize = 20;
    /**
     * 活跃时间
     */
    private static final long keepLive = 10;
    /**
     * 阻塞队列，默认队列长度为最大
     */
    private static final BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
    /**
     * 线程池执行器
     */
    private static final ThreadPoolExecutor executor;

    static {
        executor = new CustThreadPoolExecutor(
                corSize,
                maxSize,
                keepLive,
                TimeUnit.SECONDS,
                queue,
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
        logger.info("ThreadPool_executor_init_success,size:{}", corSize);
    }

    public static class CustThreadPoolExecutor extends ThreadPoolExecutor{

        public CustThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
                                      TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
                                      RejectedExecutionHandler rejectedExecutionHandler) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, rejectedExecutionHandler);
        }

        @Override
        protected void beforeExecute(Thread t, Runnable r) {
            System.out.println("线程开始执行之前");
        }

        @Override
        protected void afterExecute(Runnable r, Throwable t) {
            System.out.println("线程开始执行之后");
        }

    }

    /**
     * 无返回值线程执行
     */
    public static void putThread(Runnable runnable) {
        Map<String, String> context = MDC.getCopyOfContextMap();
        Future<?> submit = executor.submit(() -> run(runnable, context));
    }

    /**
     * 子线程加入父线程MDC属性
     */
    private static void run(Runnable runnable, Map<String, String> context) {
        // 将父线程的MDC内容传给子线程
        if (context != null) {
            try {
                MDC.setContextMap(context);
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        }
        try {
            // 执行异步操作
            runnable.run();
        } finally {
            // 清空MDC内容
            MDC.clear();
        }
    }

    /**
     * 有返回值线程执行
     */
    public static <T> Future<T> putThreadWitReturn(Callable<T> callable) throws Exception {
        Map<String, String> context = MDC.getCopyOfContextMap();
        return executor.submit(() -> run(callable, context));
    }

    /**
     * 子线程加入父线程MDC属性
     */
    private static <T> T run(Callable<T> callable, Map<String, String> context) throws Exception {
        // 将父线程的MDC内容传给子线程
        if (context != null) {
            try {
                MDC.setContextMap(context);
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        }
        T call;
        try {
            // 执行异步操作
            call = callable.call();
        } finally {
            // 清空MDC内容
            MDC.clear();
        }
        return call;
    }

    /**
     * 异步处理集合任务
     */
    public static <T> List<T> syncDeal(List<Callable<T>> callables) throws Exception {
        List<Future<T>> list = new ArrayList<>(callables.size());

        // 进行异步处理，并获取返回对象
        for(Callable<T> callable : callables){
            list.add(putThreadWitReturn(callable));
        }

        // 获取处理值
        List<T> result = new ArrayList<>(list.size());
        for (Future<T> ss : list) {
            result.add(ss.get());
        }
        return result;
    }

    public static void main(String[] args) {
        test1();
//        test2();
    }

    public static void test1(){
        try {
            MDC.put("traceId", "123");
            long start = System.currentTimeMillis();
            List<Callable<String>> callables = new ArrayList<>();
            for (int i = 0; i < 10; i++) {
                final int a = i;
                Callable<String> ss = new Callable<String>() {
                    @Override
                    public String call() throws Exception {
                        Thread.sleep(100);
                        logger.info("thread-mdc:"  + a);
                        return a + "";
                    }
                };
                callables.add(ss);
            }

            List<String> result = syncDeal(callables);

            System.out.println("test1_list:" + JSONObject.toJSONString(result));
            System.out.println("test1_cost:" + (System.currentTimeMillis() - start) / 1000 + "s");

        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            MDC.clear();
        }
    }

    public static void test2(){
        try {
            long start = System.currentTimeMillis();
            List<String> result = new ArrayList<>(10);
            for (int i = 0; i < 10; i++) {
                Thread.sleep(100);
                result.add(i + "");
            }

            System.out.println("test2_list:" + JSONObject.toJSONString(result));
            System.out.println("test2_cost:" + (System.currentTimeMillis() - start) / 1000 + "s");

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
